package com.algo.pmtong;

import java.io.IOException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.algo.panelmn.Counter;
import com.supertool.tong.merger.util.MzSequenceFileInputFormat;

/**
 * extract urls and ips of each uuid
 * author: lujianfeng
 *
 */
public class GetUuidUrlIp extends Configured implements Tool{

	public static class ExtractFeature extends Mapper<LongWritable, Text, Text, Text>{
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
			String line = value.toString();
			String[] items = line.split("[=\\^]");
			Map map = new HashMap();
			for(int i = 1; i < items.length; i = i + 2)
				map.put(items[i-1], items[i]);
			String plt = (String) map.get("plt");
			if(plt.equals("0")){
				StringBuilder urlIp = new StringBuilder();
				if(map.containsKey("c")){
					urlIp.append(URLDecoder.decode(((String) map.get("c")).replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8"));
				}
				else if(map.containsKey("pr")){
					urlIp.append(URLDecoder.decode(((String) map.get("pr")).replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8"));
				}
				urlIp.append("^").append(map.get("ip"));
				context.write(new Text((String) map.get("uuid")), new Text(urlIp.toString()));
			}
		}
	}
	public static class Mappersns extends Mapper<LongWritable, Text, Text, Text>{
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] sns = value.toString().split("\\^");
			if(sns.length == 4)
				context.write(new Text(sns[0]), new Text(sns[1]));
			else	System.out.println("less sns element");
		}
	}
	public static class SNSTong extends Reducer<Text, Text, Text, Text>{
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException{
			Iterator it = values.iterator();
			Map<String, Counter> urlPv = new HashMap<String, Counter>();
			Map<String, Counter> ipPv = new HashMap<String, Counter>();
			List sns = new ArrayList();
			while(it.hasNext()){
				String[] value = ((Text) it.next()).toString().split("\\^");
				if(value.length == 1){
					sns.add(value[0]);
				}
				else if(value.length  == 2){
					String url;
					if(value[0].startsWith("http://"))
						url = value[0].substring(7);
					else	url = value[0];
					String[] segments = url.split("\\/");
					if(segments.length <= 0) continue;
					String webName = segments[0];
					if(urlPv.containsKey(webName)){
						Counter pv = (Counter)urlPv.get(webName);
						pv.num++;
					}
					else	urlPv.put(webName, new Counter(1));
					String ip = value[1];
					if(ipPv.containsKey(ip)){
						Counter pv = (Counter) ipPv.get(ip);
						pv.num++;
					}
					else	ipPv.put(ip, new Counter(1));
				}
			}
			//output uuid feature corresponding to one snsid
			if(sns.size() == 1){
				StringBuilder urls = new StringBuilder(key.toString());
				urls.append("^");
				for(Map.Entry<String, Counter> entry : urlPv.entrySet()){
					urls.append(entry.getKey()).append("=").append(entry.getValue().num).append("#");
				}
				urls.deleteCharAt(urls.length() - 1).append("^");
				for(Map.Entry<String, Counter> entry : ipPv.entrySet()){
					urls.append(entry.getKey()).append("=").append(entry.getValue().num).append("#");
				}
				urls.deleteCharAt(urls.length() - 1);
				context.write(new Text((String) sns.get(0)), new Text(urls.toString()));
			}
		}
	}
	public int getMonthDay(int year, int month){
		if(month > 12 || month < 1) return 0;
		if(month%2 == 1 && month <= 7) return 31;
		else if(month%2 == 0 && month >= 8) return 31;
		else if(month != 2) return 30;
		else{
			if((year%4 == 0 && year%100 != 0) || year%400 == 0) return 29;
			else return 28;
		}
	}
	void addPathToJob(String fromTime, String toTime, Job job, String rootDir, Configuration conf, 
			Class inputFormatClass, Class mapperClass, boolean isLog) throws IOException{
		int reducenum = 0;
		int fromyear = Integer.parseInt(fromTime.substring(0, 4));
        int toyear = Integer.parseInt(toTime.substring(0, 4));
        int frommonth = Integer.parseInt(fromTime.substring(4, 6));
        int tomonth = Integer.parseInt(toTime.substring(4, 6));
        int fromday = Integer.parseInt(fromTime.substring(6));
        int today = Integer.parseInt(toTime.substring(6));
        FileSystem hdfs = FileSystem.get(conf);
        for(int i = fromyear; i <= toyear; i++){
        	for(int j = frommonth; j <= tomonth; j++){
        		int lastday = getMonthDay(i, j);
        		int firstday = 1;
        		if(i == fromyear && j == frommonth)
        			firstday = fromday;
        		if(i == toyear && j == tomonth)
        			lastday = today;
        		String month = j < 10 ? ("0" + j) : ("" + j);
        		for(int k = firstday; k <= lastday; k++){
        			String day = k < 10 ? ("0" + k) : ("" + k);
        			Path p1;
        			if(isLog){
        				p1 = new Path(rootDir + "/" + i + month + day);
        				FileStatus[] files = hdfs.listStatus(p1);
	        	        for(int l = 0;l < files.length;l++){
	        	        	Path file = files[l].getPath();
	        	        	if(file.getName().indexOf("campaign") >= 0)
	        	        		MultipleInputs.addInputPath(job, file, inputFormatClass, mapperClass);
	        	        }
	        	        reducenum++;
        			}
        			else{
        				p1 = new Path(rootDir+ "/" + i + month + "/" + i + month + day);
        				MultipleInputs.addInputPath(job, p1, inputFormatClass, mapperClass);
        			}
        		}
        	}
        }
        if(isLog)	job.setNumReduceTasks(5 + 2*reducenum);
	}
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		Job job = Job.getInstance(conf, "uuid feature");
		job.setJarByClass(GetUuidUrlIp.class);
		
		//add path to job according to start and end time
		addPathToJob(args[2], args[3], job, args[0], conf, MzSequenceFileInputFormat.class, ExtractFeature.class, true);
		addPathToJob(args[2], args[3], job, args[1], conf, TextInputFormat.class, Mappersns.class, false);
		
		FileOutputFormat.setOutputPath(job, new Path(args[4]));
		job.setReducerClass(SNSTong.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
		
		return job.waitForCompletion(true) ? 1 : 0;
	}
	public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new GetUuidUrlIp(), args);
        System.exit(res);
    }

}
